package com.hudsonmq.spring.impl.consume.management;


import com.hudson.codec.domain.MessageExt;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.util.CollectionUtil;
import com.hudson.core.util.SnowflakeIdGenerator;
import com.hudsonmq.spring.impl.consume.HudsonMQConsumer;
import com.hudsonmq.spring.impl.consume.HudsonMQHandler;
import com.hudsonmq.spring.utils.NettyUtil;
import io.netty.channel.Channel;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: 渔小鲤
 * @DateTime: 2025/7/12 18:00
 **/
@Component
@ConditionalOnProperty(prefix = "hudson",name = "consume-group")
public class ConsumerHandlerManagement implements InitializingBean {

    private static final Logger logger = LoggerFactory.getLogger(ConsumerHandlerManagement.class);
    static ConcurrentHashMap<String, HudsonMQHandler> HANDLER_MAP = new ConcurrentHashMap<>();
    private static AtomicInteger threadIndex = new AtomicInteger();
    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(
            20,
            40,
            30,
            TimeUnit.SECONDS,
            new LinkedBlockingDeque<>(),
            new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "Custom-Executor-" + threadIndex.incrementAndGet());
                }
            });

    @Resource
    private List<HudsonMQHandler> handlerList;



    public void onMessage(Channel channel, List<MessageExt> messageExtList) {
        if (messageExtList.isEmpty()) {
            return;
        }
        HudsonMQHandler hudsonMQHandler = null;
        MessageExt messageExts = messageExtList.get(0);
        long startTime = System.currentTimeMillis();

        List<MessageExt> retryMessage = new ArrayList<>();
        try {
            hudsonMQHandler = HANDLER_MAP.get(messageExts.getTopic());
            for (MessageExt messageExt : messageExtList) {
                logger.debug("开始消费,消费信息:{}", messageExt.toString());
                try {

                    hudsonMQHandler.onMessage(messageExt.getBody());
                } catch (Exception e) {
                    logger.error("消费异常:{}", e.getMessage());
                    messageExt.setReconsumeTimes(messageExt.getReconsumeTimes() + 1);
                    retryMessage.add(messageExt);
                }
            }
            logger.debug("队列 {} 消费成功！", messageExts.getTopic() + messageExts.getQueueId());
        } catch (Exception e) {
            logger.debug("消费失败:");
            hudsonMQHandler.retry(messageExtList);
        } finally {
            if (CollectionUtil.noEmpty(retryMessage)) {
                NettyUtil.write(channel, RemotingCommand.builder()
                        .type(MessageTypeEnum.CONSUMER_RETRY_MESSAGE.getType())
                        .opaque(SnowflakeIdGenerator.DEFAULT.nextId())
                        .body(MessageExt.list2Bytes(retryMessage))
                        .build());
            }

            PushMessageMonitor.decrementQueueSize(messageExts.getTopic(), messageExts.getQueueId(), messageExtList.size());
            long consumerTime = System.currentTimeMillis() - startTime;
            PushMessageMonitor.changePullMessageSize(messageExts.getTopic(), messageExts.getQueueId(), consumerTime);
        }
    }

    public void exec(Runnable runnable) {
        executor.execute(runnable);
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        for (HudsonMQHandler hudsonMQHandler : handlerList) {
            HudsonMQConsumer annotation = hudsonMQHandler.getClass().getAnnotation(HudsonMQConsumer.class);
            if (Objects.nonNull(annotation)) {
                HANDLER_MAP.put(annotation.topic(), hudsonMQHandler);
            }
        }
    }
}
